package xin.petstore.framework.canal;

import com.alibaba.otter.canal.client.CanalConnector;
import com.alibaba.otter.canal.client.CanalConnectors;
import com.alibaba.otter.canal.protocol.CanalEntry.Column;
import com.alibaba.otter.canal.protocol.CanalEntry.Entry;
import com.alibaba.otter.canal.protocol.CanalEntry.EntryType;
import com.alibaba.otter.canal.protocol.CanalEntry.EventType;
import com.alibaba.otter.canal.protocol.CanalEntry.Header;
import com.alibaba.otter.canal.protocol.CanalEntry.RowChange;
import com.alibaba.otter.canal.protocol.CanalEntry.RowData;
import com.alibaba.otter.canal.protocol.Message;
import com.googlecode.asyn4j.core.WorkWeight;
import com.googlecode.asyn4j.core.callback.AsynCallBack;
import com.googlecode.asyn4j.service.AsynService;

import java.net.InetSocketAddress;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

import org.apache.commons.lang3.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.slf4j.MDC;
import org.springframework.beans.BeansException;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.context.ApplicationContext;
import org.springframework.context.ApplicationContextAware;
import org.springframework.util.Assert;

public class CanalDataListener
        implements InitializingBean, ApplicationContextAware {
    private static final Logger logger = LoggerFactory.getLogger(CanalDataListener.class);
    private volatile boolean running = false;
    private Thread.UncaughtExceptionHandler handler = new Thread.UncaughtExceptionHandler() {
        public void uncaughtException(Thread t, Throwable e) {
            CanalDataListener.logger.error("parse events has an error", e);
        }
    };

    private boolean isAsync = false;
    private ApplicationContext applicationContext;
    private int batchSize = 5120;
    private Thread thread = null;
    private CanalConnector connector;
    private AsynService asynService;
    private String host;
    private int port;
    private String destination;
    private String username;
    private String password;
    private String zkServers;
    private String addresses;
    private Map<String, String> tableToService = new HashMap();

    protected void start() {
        Assert.notNull(this.connector, "connector is null");
        this.thread = new Thread(new Runnable() {
            public void run() {
                CanalDataListener.this.process();
            }
        });
        this.thread.setUncaughtExceptionHandler(this.handler);
        this.thread.start();
        this.running = true;
        logger.info("----------start canal-->running:" + this.running);
    }

    protected void stop() {
        if (!this.running) {
            return;
        }
        this.running = false;
        if (this.thread != null) {
            try {
                this.thread.join();
            } catch (InterruptedException e) {
                logger.error("stop events has an error", e);
            }
        }
        MDC.remove("destination");
    }

    protected void process() {
        while (this.running)
            try {
                MDC.put("destination", this.destination);
                this.connector.connect();
                this.connector.subscribe();
                while (this.running) {
                    long batchId = 0L;
                    try {
                        Message message = this.connector.getWithoutAck(this.batchSize);
                        batchId = message.getId();
                        int size = message.getEntries().size();
                        if ((batchId == -1L) || (size == 0))
                            try {
                                Thread.sleep(1000L);
                            } catch (InterruptedException e) {
                                e.printStackTrace();
                            }
                        else {
                            try {
                                processEntry(message.getEntries());
                            } catch (Exception e) {
                                if (logger.isErrorEnabled())
                                    logger.error("processEntry error!", e);
                            }
                        }
                        this.connector.ack(batchId);
                    } catch (Exception e) {
                        if (logger.isErrorEnabled())
                            logger.error("process error!", e);
                        if (batchId > 0L)
                            this.connector.rollback(batchId);
                    }
                }
            } catch (Exception e) {
                if (logger.isErrorEnabled())
                    logger.error("process error!", e);
            } finally {
                this.connector.disconnect();
                MDC.remove("destination");
            }
    }

    protected void processEntry(List<CanalEntry.Entry> entrys) {
        for (CanalEntry.Entry entry : entrys) {
            if ((entry.getEntryType() == CanalEntry.EntryType.TRANSACTIONBEGIN) ||
                    (entry.getEntryType() == CanalEntry.EntryType.TRANSACTIONEND)) {
                continue;
            }
            if (entry.getEntryType() == CanalEntry.EntryType.ROWDATA) {
                CanalEntry.RowChange rowChage = null;
                try {
                    rowChage = CanalEntry.RowChange.parseFrom(entry.getStoreValue());
                } catch (Exception e) {
                    throw new RuntimeException("parse event has an error , data:" + entry.toString(), e);
                }
                CanalEntry.EventType eventType = rowChage.getEventType();
                if ((eventType == CanalEntry.EventType.QUERY) || (rowChage.getIsDdl())) {
                    if (logger.isDebugEnabled()) {
                        logger.debug(" sql ----> " + rowChage.getSql());
                    }
                } else
                    for (CanalEntry.RowData rowData : rowChage.getRowDatasList()) {
                        String tableName = entry.getHeader().getTableName();
                        String schemaName = entry.getHeader().getSchemaName();
                        String serviceName = (String) this.tableToService.get(schemaName + "." + tableName);

                        if (StringUtils.isNotEmpty(serviceName)) {
                            List objs = new ArrayList();
                            Map pars = new HashMap();
                            if (logger.isDebugEnabled()) {
                                logger.debug("SchemaName:{} eventType:{} tableName:{} serviceName:{}", new String[]{schemaName, entry.getEntryType().toString(), tableName, serviceName});
                            }
                            if (eventType == CanalEntry.EventType.DELETE) {
                                processDataObject(objs, pars, rowData.getBeforeColumnsList());
                                call(serviceName, "processDel", objs, pars, tableName);
                            } else if (eventType == CanalEntry.EventType.INSERT) {
                                processDataObject(objs, pars, rowData.getAfterColumnsList());
                                call(serviceName, "processInsert", objs, pars, tableName);
                            } else if (eventType == CanalEntry.EventType.UPDATE) {
                                processDataObject(objs, pars, rowData.getAfterColumnsList());
                                call(serviceName, "processUpdate", objs, pars, tableName);
                            }
                        }
                    }
            }
        }
    }

    private void call(String serviceName, String serviceMethodName, List<DataObject> dataObjects, Map<String, String> dataMap, String tableName) {
        try {
            if (this.isAsync) {
                this.asynService.addWorkWithSpring(serviceName, serviceMethodName, new Object[]{dataObjects, dataMap, tableName}, new MyAsynCallBack(tableName), WorkWeight.MIDDLE);
            } else {
                DataProcess dp = (DataProcess) this.applicationContext.getBean(serviceName);
                if (serviceMethodName.equalsIgnoreCase("processDel"))
                    dp.processDel(dataObjects, dataMap, tableName);
                else if (serviceMethodName.equalsIgnoreCase("processInsert"))
                    dp.processInsert(dataObjects, dataMap, tableName);
                else if (serviceMethodName.equalsIgnoreCase("processUpdate"))
                    dp.processUpdate(dataObjects, dataMap, tableName);
            }
        } catch (Exception e) {
            if (logger.isErrorEnabled())
                logger.error("call error\n", e);
        }
    }

    private void processDataObject(List<DataObject> objs, Map<String, String> pars, List<CanalEntry.Column> columns) {
        for (CanalEntry.Column c : columns) {
            DataObject d = new DataObject();
            d.setMysqlType(c.getMysqlType());
            d.setNull(c.getIsNull());
            d.setKey(c.getIsKey());
            d.setUpdate(c.getUpdated());
            d.setName(c.getName());
            d.setValue(c.getValue());
            objs.add(d);
            pars.put(c.getName(), c.getValue());
        }
    }

    public void afterPropertiesSet() throws Exception {
        Assert.notNull(this.username, "username is null");
        Assert.notNull(this.password, "password is null");
        Assert.notNull(this.destination, "destination is null");
        if (this.connector == null) {
            if (this.zkServers != null) {
                this.connector = CanalConnectors.newClusterConnector(this.zkServers, this.destination, this.username, this.password);
                logger.info("----------start host-->zkServers:" + this.zkServers);
            } else if ((this.host != null) && (this.port > 0)) {
                this.connector = CanalConnectors.newSingleConnector(
                        new InetSocketAddress(this.host, this.port), this.destination, this.username, this.password);
                logger.info("----------start host-->host:" + this.host + " port:" + this.port);
            } else if (StringUtils.isNotEmpty(this.addresses)) {
                List ads = new ArrayList();
                for (String en : this.addresses.split(",")) {
                    String[] adss = en.split(":");
                    ads.add(new InetSocketAddress(adss[0], Integer.parseInt(adss[1])));
                }
                this.connector = CanalConnectors.newClusterConnector(ads, this.destination, this.username, this.password);
                logger.info("----------start host-->addresses:" + this.addresses);
            }
        }
        start();
    }

    public CanalConnector getConnector() {
        return this.connector;
    }

    public void setConnector(CanalConnector connector) {
        this.connector = connector;
    }

    public AsynService getAsynService() {
        return this.asynService;
    }

    public void setAsynService(AsynService asynService) {
        this.asynService = asynService;
    }

    public String getHost() {
        return this.host;
    }

    public void setHost(String host) {
        this.host = host;
    }

    public int getPort() {
        return this.port;
    }

    public void setPort(int port) {
        this.port = port;
    }

    public String getDestination() {
        return this.destination;
    }

    public void setDestination(String destination) {
        this.destination = destination;
    }

    public String getUsername() {
        return this.username;
    }

    public void setUsername(String username) {
        this.username = username;
    }

    public String getPassword() {
        return this.password;
    }

    public void setPassword(String password) {
        this.password = password;
    }

    public String getZkServers() {
        return this.zkServers;
    }

    public void setZkServers(String zkServers) {
        this.zkServers = zkServers;
    }

    public String getAddresses() {
        return this.addresses;
    }

    public void setAddresses(String addresses) {
        this.addresses = addresses;
    }

    public Map<String, String> getTableToService() {
        return this.tableToService;
    }

    public void setTableToService(Map<String, String> tableToService) {
        this.tableToService = tableToService;
    }

    public boolean isAsync() {
        return this.isAsync;
    }

    public void setAsync(boolean isAsync) {
        this.isAsync = isAsync;
    }

    public void setApplicationContext(ApplicationContext applicationContext)
            throws BeansException {
        this.applicationContext = applicationContext;
    }

    class MyAsynCallBack extends AsynCallBack {
        private String tableName;

        public MyAsynCallBack(String tableName) {
            this.tableName = tableName;
        }

        public void doNotify() {
            if (CanalDataListener.logger.isDebugEnabled())
                CanalDataListener.logger.debug("table {} process fining", this.tableName);
        }
    }
}
